//
// Created by dave on 18-7-31.
//

#pragma once
#include <atomic>
#include <vector>
#include <ctime>
#include <cassandra.h>
#include <glog/logging.h>
#include "location.h"

class location_dao {
public:
    location_dao(CassSession* session) : session_(session) { }

    void insert(const std::vector<location>& vec) {
        // clock_t t = clock();
        CassFuture* future = cass_session_prepare(session_, "INSERT INTO iot.location (sim,alarmFlag,flag,located,lat,lng,height,speed,direction,\"uuid\") "
            "VALUES (?,?,?,?,?,?,?,?,?,?)");
        cass_future_wait(future);
        CassError rc0 = cass_future_error_code(future);
        if (rc0 != CASS_OK) {
            const char* message;
            size_t message_length;
            cass_future_error_message(future, &message, &message_length);
            LOG(INFO) << "Batch result: " << message;
            cass_future_free(future);
            return;
        }

        const CassPrepared* prepared = cass_future_get_prepared(future);
        cass_future_free(future);

        CassBatch* batch = cass_batch_new(CASS_BATCH_TYPE_LOGGED);
        for (int i = 0; i < vec.size(); i++) {
            CassStatement* statement = cass_prepared_bind(prepared);
            cass_statement_bind_string(statement, 0, vec[i].get_sim().c_str());
            cass_statement_bind_int32(statement, 1, vec[i].get_alarm());
            cass_statement_bind_int32(statement, 2, vec[i].get_flag());
            cass_statement_bind_bool(statement, 3, vec[i].is_located() ? cass_true : cass_false);
            cass_statement_bind_double(statement, 4, vec[i].get_lat());
            cass_statement_bind_double(statement, 5, vec[i].get_lng());
            cass_statement_bind_int32(statement, 6, vec[i].get_height());
            cass_statement_bind_int32(statement, 7, vec[i].get_speed());
            cass_statement_bind_int32(statement, 8, vec[i].get_direction());
            cass_statement_bind_uuid(statement, 9, vec[i].get_uuid());
            cass_batch_add_statement(batch, statement);
            cass_statement_free(statement);
        }

        CassFuture* batch_future = cass_session_execute_batch(session_, batch);
        cass_batch_free(batch);
        cass_prepared_free(prepared);

        CassError rc = cass_future_error_code(batch_future);
        if (rc != CASS_OK) {
            const char* message;
            size_t message_length;
            cass_future_error_message(batch_future, &message, &message_length);
            LOG(INFO) << "Batch result: " << message;
            cass_future_free(batch_future);
            return;
        }
        cass_future_free(batch_future);
        // LOG(INFO) << "插入" << vec.size() << ", " << ((clock() - t) / (CLOCKS_PER_SEC / 1000)) << "ms";
    }

private:
    CassSession* session_;
};